package com.clw.dwd;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.clw.cdc.CustomerDeserialization;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*@SpringBootApplication(exclude = GsonAutoConfiguration.class)
@Log4j2*/
public class DynamicStreamByCdc // implements CommandLineRunner
{
    public static void main(String[] args) throws Exception {
       // SpringApplication.run(DynamicStreamByCdc.class,args);
        cdc();
    }

  //  @Override
    public void run(String... args) throws Exception {


    }
    public static void cdc() throws Exception {
        // 2. 使用CDC 获取数据
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DebeziumSourceFunction<String> sourceFunction =  MySQLSource.<String>builder()
                .hostname("hadoop-single")
                .port(3306)
                .username("root")
                .password("Abcd1234!")
                .databaseList("gmall")
                .tableList("gmall.*")   //如果不添加该参数,则消费指定数据库中所有表的数据.如果指定,指定方式为db.table
                .deserializer(new CustomerDeserialization())
                .startupOptions(StartupOptions.initial())
                .build();
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
        streamSource.print();
        env.execute();
    }
}
